package task1

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CityViewCount {
  def main(args: Array[String]): Unit = {
    // 仍需完善

//    Logger.getLogger("org").setLevel(Level.WARN)
    val conf = new SparkConf()
      .setAppName(s"${this.getClass.getCanonicalName}")
      //      .setMaster("spark://linux121:7077")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")

    // 读取数据（本地）
    val logRDD = sc.textFile("file:///D:\\projects\\spark_homework\\data\\http.log", 10)
    val ipRDD = sc.textFile("file:///D:\\projects\\spark_homework\\data\\ip.dat", 10)
    // hdfs
    //    val logRDD = sc.textFile("hdfs://linux121:9000/user/data/spark/http.log", 10)
    //    val ipRDD = sc.textFile("hdfs://linux121:9000//user/data/spark/ip.dat")


    //    // 测试
    //    val logRDD = sc.makeRDD(Seq("20090121000132095572000|125.213.100.123|show.51.com|/shoplist.php?phpfile=shoplist2.php&style=1&sex=137|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; Mozilla/4.0(Compatible Mozilla/4.0(Compatible-EmbeddedWB 14.59 http://bsalsa.com/ EmbeddedWB- 14.59  from: http://bsalsa.com/ )|http://show.51.com/main.php|",
    //    "20090121000132581311000|115.120.36.118|tj.tt98.com|/tj.htm|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TheWorld)|http://www.tt98.com/|"))
    //    val ipRDD = sc.makeRDD(Seq("1.0.1.0|1.0.3.255|16777472|16778239|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302", "1.0.8.0|1.0.15.255|16779264|16781311|亚洲|中国|广东|广州||电信|440100|China|CN|113.280637|23.125178"))


    // 思路：将ipRDD按照ip排序，将这个RDD广播出去，在每个executor上，通过二分查找寻找logRDD中ip对应的城市

    // 提取ipRDD中的ip段和城市
    val ipRange: RDD[(Long, Long, String)] = ipRDD.map(x => {
      val fields = x.split("\\|")
      val start = fields(0).split(".").zipWithIndex.map{case (x, i) => x.toLong * math.pow(/*256*/1000, 3 - i).toLong}.sum
      val end = fields(1).split(".").zipWithIndex.map{case (x, i) => x.toLong * math.pow(/*256*/1000, 3 - i).toLong}.sum
      val city = fields(7)
      (start, end, city)
    })

    // 将该变量按照其实ip排序后广播
    val bc = sc.broadcast(ipRange.collect.toList.sortBy(_._1))

    // 提取logRDD中的ip，形成 (ip, log)
//    println(logRDD.getNumPartitions)
    val changedLog: RDD[(String, String)] = logRDD.map(x => {
      val log = x.split("\\|")
      (log(1).split(".").zipWithIndex.map{case (x, i) => x.toLong * math.pow(/*256*/1000, 3 - i).toLong}.sum, log)
    })
//    println(changedLog.getNumPartitions)
      .mapPartitions(arrs => {
        val ipRange = bc.value
        val tuples: Iterator[(String, String)] = arrs.map { case (ip, log) =>
          val idx: Option[Int] = lookForIp(ip, ipRange.length / 2, ipRange)
          var addr: String = "未知"
          if (idx.isDefined) addr = ipRange(idx.get)._3
          log(1) = addr
          (addr, log.mkString("|"))
        }
        tuples
      })
    // 打印部分结果
    println("==================部分结果展示===================")
    changedLog.map(_._2).take(20).foreach(println)

    val result = changedLog.countByKey()
    println("==============各城市访问量统计===================")
    result.foreach(println)

    Thread.sleep(1000000)

    sc.stop()
  }


  @scala.annotation.tailrec
  def lookForIp(ip: Long, midInx: Int, ipRange: List[(Long, Long, String)]): Option[Int] = {
    val start = ipRange(midInx)._1
    val end = ipRange(midInx)._2
    if (ip >= start && ip <= end) Some(midInx)
    else if (midInx == 0 || midInx == ipRange.size - 1) None
    else if (ip < start) lookForIp(ip, midInx / 2, ipRange)
    else lookForIp(ip, (ipRange.size - midInx) / 2, ipRange)
  }
}
